#!/bin/bash
# 获取ETL时间  今天
dt=`date +"%Y-%m-%d"`
# 定义需要全量抽取的 表名  数据集
arr=(AML_ACCOUNT_MASTER AML_ALERT AML_CUSTOMER_MASTER AML_MONITORING_RULE AML_SCREENING_RESULT AML_SUSPICIOUS_TXN_REPORT AML_TRANSACTION_DETAIL AML_UBO_INFO AML_WATCHLIST_DETAIL AML_WATCHLIST_MASTER)

# 循环数值
for i in ${arr[@]}; do
hive_tb="ODS_$i"
  sqoop import \
  --connect jdbc:oracle:thin:@192.168.65.41:1521/orcl \
  --username FXQ \
  --password 123456 \
  --table $i \
  --hcatalog-database ods \
  --hcatalog-table $hive_tb \
  --hive-partition-key etl_date \
  --hive-partition-value $dt \
  -m 1
done


#!/bin/bash
# 配置参数
ORACLE_CONN="jdbc:oracle:thin:@192.168.65.41:1521/orcl"
USER="FXQ"
PASS="123456"
DT=$(date +%Y%m%d)
# 定义需要全量快照的 表名  数据集
arr=(AML_CUSTOMER_MASTER)


# 表映射数组（Oracle表名:Hive表名）
declare -A TABLES=(
  ["AML_CUSTOMER_MASTER"]="ods_aml_customer_master"
  )

for i in "${!TABLES[@]}"; do
  hive_tb=${TABLES[$i]}
  echo "开始同步表: $i -> $hive_tb"

  sqoop import \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --table $i \
    --hcatalog-database ods \
    --hcatalog-table $hive_tb \
    --hive-partition-key etl_date \
    --hive-partition-value $DT \
    --direct \
    --fetch-size 2000 \
    --compress \
    --compression-codec snappy \
    --hive-overwrite \
    -m 1

  if [ $? -eq 0 ]; then
    echo "$i 表同步成功"
  else
    echo "$i 表同步失败"
    exit 1
  fi
done

#!/bin/bash
# 增量导入配置
ORACLE_CONN="jdbc:oracle:thin:@192.168.65.41:1521/orcl"
USER="FXQ"
PASS="123456"
DT=$(date +%Y%m%d)
# 定义需要全量抽取的 表名  数据集
arr=(AML_ACCOUNT_MASTER AML_ALERT AML_CUSTOMER_MASTER AML_MONITORING_RULE AML_SCREENING_RESULT AML_SUSPICIOUS_TXN_REPORT AML_TRANSACTION_DETAIL AML_UBO_INFO AML_WATCHLIST_DETAIL AML_WATCHLIST_MASTER)

LAST_VALUE_FILE="/path/to/last_value_record.txt"  # 存储上次导入的last-value

# 表映射配置
declare -A TABLES=(
  ["SRC_TABLE1"]="TARGET_TABLE1"
  ["SRC_TABLE2"]="TARGET_TABLE2"
)

# 模式1：基于自增ID的append模式（适合仅追加场景）
incremental_append() {
  local src_tb=$1
  local hive_tb=$2
  local check_col="ID"  # 需替换为实际自增列

  # 读取上次记录的last-value（首次运行时默认为0）
  local last_value=$(cat ${LAST_VALUE_FILE}_${src_tb} 2>/dev/null || echo 0)

  sqoop import \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --table $src_tb \
    --hcatalog-database ods \
    --hcatalog-table $hive_tb \
    --hive-partition-key etl_date \
    --hive-partition-value $DT \
    --incremental append \
    --check-column $check_col \
    --last-value $last_value \
    --direct \
    --compress \
    -m 1

  # 更新last-value记录
  sqoop eval \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --query "SELECT MAX($check_col) FROM $src_tb" \
    >> ${LAST_VALUE_FILE}_${src_tb}
}

# 模式2：基于时间戳的lastmodified模式（适合有更新的场景）
incremental_lastmodified() {
  local src_tb=$1
  local hive_tb=$2
  local time_col="UPDATE_TIME"  # 需替换为实际时间戳列

  # 获取上次导入时间（首次运行时默认为前一天）
  local last_time=$(cat ${LAST_VALUE_FILE}_${src_tb} 2>/dev/null || \
                   date -d "1 day ago" +"%Y-%m-%d %H:%M:%S")

  sqoop import \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --table $src_tb \
    --hcatalog-database ods \
    --hcatalog-table $hive_tb \
    --hive-partition-key etl_date \
    --hive-partition-value $DT \
    --incremental lastmodified \
    --check-column $time_col \
    --last-value "$last_time" \
    --merge-key "ID"  # 需指定主键列用于合并更新
    --direct \
    --compress \
    -m 1

  # 更新last-value记录
  sqoop eval \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --query "SELECT MAX($time_col) FROM $src_tb" \
    >> ${LAST_VALUE_FILE}_${src_tb}
}

# 主执行逻辑
for src_tb in "${!TABLES[@]}"; do
  hive_tb=${TABLES[$src_tb]}
  echo "开始增量同步: $src_tb -> $hive_tb"

  # 根据表特性选择增量模式（示例自动判断逻辑需根据实际情况调整）
  if [[ $src_tb =~ "HIS_" ]]; then
    incremental_append $src_tb $hive_tb
  else
    incremental_lastmodified $src_tb $hive_tb
  fi
done


#!/bin/bash
# 全量覆盖导入配置
ORACLE_CONN="jdbc:oracle:thin:@192.168.65.41:1521/orcl"
USER="FXQ"
PASS="123456"
DT=$(date +%Y%m%d)
LOG_DIR="/var/log/sqoop_import"
mkdir -p $LOG_DIR

# 表映射配置（Oracle表名:Hive表名）
declare -A TABLES=(
  ["CUSTOMER"]="ods_customer"
  ["TRANSACTION"]="ods_transaction"
  ["PRODUCT"]="ods_product"
)

# 执行全量导入
for src_tb in "${!TABLES[@]}"; do
  hive_tb=${TABLES[$src_tb]}
  log_file="${LOG_DIR}/${src_tb}_${DT}.log"

  echo "$(date '+%Y-%m-%d %H:%M:%S') 开始导入表: $src_tb -> $hive_tb" | tee -a $log_file

  sqoop import \
    --connect $ORACLE_CONN \
    --username $USER \
    --password $PASS \
    --table $src_tb \
    --hcatalog-database ods \
    --hcatalog-table $hive_tb \
    --hive-partition-key etl_date \
    --hive-partition-value $DT \
    --hive-overwrite \
    --direct \
    --compress \
    --compression-codec snappy \
    --fields-terminated-by '\001' \
    --lines-terminated-by '\n' \
    --null-string '\\N' \
    --null-non-string '\\N' \
    -m 4 \
    --split-by "ROWNUM" \
    --outdir /tmp/sqoop_codegen \
    2>&1 | tee -a $log_file

  if [ ${PIPESTATUS[0]} -eq 0 ]; then
    echo "$(date '+%Y-%m-%d %H:%M:%S') 表 $src_tb 导入成功" | tee -a $log_file
  else
    echo "$(date '+%Y-%m-%d %H:%M:%S') 表 $src_tb 导入失败" | tee -a $log_file
    exit 1
  fi
done
